package com.shujia.dim

import com.shujia.util.FlinkTool

object DimKafkaMysqlSku extends FlinkTool{
  /**
    * 需要在子类实现抽象方法
    *
    */
  override def run(args: Array[String]): Unit = {
    table.executeSql(
      """
        |
        |insert into gma_dim.dim_kafka_mysql_sku
        |select
        |a.id,
        |a.spu_id,
        |a.price,
        |a.sku_name,
        |a.sku_desc,
        |a.weight,
        |b.tm_name,
        |c.name as category3_name,
        |d.name as category2_name,
        |e.name as category1_name,
        |a.sku_default_img,
        |a.create_time
        |from
        |gma_ods.ods_mysql_kafka_sku_info /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as a
        |join
        |gma_ods.ods_mysql_kafka_base_trademark /*+ OPTIONS('scan.startup.mode'='earliest-offset') */  as b
        |on a.tm_id = cast(b.tm_id as bigint)
        |join
        |gma_ods.ods_mysql_kafka_base_category3 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as c
        |on a.category3_id = c.id
        |join
        |gma_ods.ods_mysql_kafka_base_category2 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as d
        |on c.category2_id = d.id
        |join
        |gma_ods.ods_mysql_kafka_base_category1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as e
        |on d.category1_id = e.id
        |
      """.stripMargin)
  }

}
